package org.apache.ocean.nio;

import java.io.IOException;
import java.io.InterruptedIOException;
import java.net.InetSocketAddress;
import java.net.URL;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
import java.util.logging.Level;
import java.util.logging.Logger;

import org.apache.commons.httpclient.HttpURL;
import org.apache.commons.httpclient.StatusLine;
import org.apache.commons.lang.StringUtils;
import org.apache.http.Header;
import org.apache.http.HttpEntity;
import org.apache.http.HttpException;
import org.apache.http.HttpRequest;
import org.apache.http.HttpResponse;
import org.apache.http.impl.DefaultConnectionReuseStrategy;
import org.apache.http.impl.nio.DefaultClientIOEventDispatch;
import org.apache.http.impl.nio.reactor.DefaultConnectingIOReactor;
import org.apache.http.message.BasicHttpRequest;
import org.apache.http.nio.NHttpConnection;
import org.apache.http.nio.protocol.BufferingHttpClientHandler;
import org.apache.http.nio.protocol.EventListener;
import org.apache.http.nio.protocol.HttpRequestExecutionHandler;
import org.apache.http.nio.reactor.ConnectingIOReactor;
import org.apache.http.nio.reactor.IOEventDispatch;
import org.apache.http.nio.reactor.SessionRequest;
import org.apache.http.nio.reactor.SessionRequestCallback;
import org.apache.http.params.BasicHttpParams;
import org.apache.http.params.CoreConnectionPNames;
import org.apache.http.params.CoreProtocolPNames;
import org.apache.http.params.HttpParams;
import org.apache.http.protocol.BasicHttpProcessor;
import org.apache.http.protocol.HttpContext;
import org.apache.http.protocol.RequestConnControl;
import org.apache.http.protocol.RequestContent;
import org.apache.http.protocol.RequestExpectContinue;
import org.apache.http.protocol.RequestTargetHost;
import org.apache.http.protocol.RequestUserAgent;
import org.apache.http.util.EntityUtils;
import org.apache.ocean.NamedList;
import org.apache.ocean.TimeoutState;
import org.apache.ocean.OceanXML.RequestException;

public class NioHttpClient {
	public static Logger log = Logger.getLogger(NioHttpClient.class.getName());
	public final static String COMPLETED = "completed";
	public final static String FAILED = "failed";
	public final static String TIMEOUT = "timeout";
	public final static String CANCELLED = "cancelled";
	HttpParams httpParams;
	ConnectingIOReactor ioReactor;
	Thread thread;
	BasicHttpProcessor basicHttpProcessor;
	BufferingHttpClientHandler clientHandler;
	IOEventDispatch ioEventDispatch;

	public static void main(String[] args) {
		try {
			NioHttpClient nioHttpClient = new NioHttpClient(2);
			List<org.apache.ocean.HttpRequest> httpRequests = new ArrayList<org.apache.ocean.HttpRequest>();
			// httpRequests.add(new org.apache.ocean.HttpRequest("1", new
			// URL("http://search.aol.com/aol/search"), new
			// HttpParameters("invocationType",
			// "topsearchbox.webhome", "query", "shark attack")));
			httpRequests.add(new org.apache.ocean.HttpRequest("2", new URL("http://www.google.com")));
			//httpRequests.add(new org.apache.ocean.HttpRequest("3", new URL("http://www.jasdfsdfsd.com")));
			TimeoutState timeoutState = new TimeoutState(1000 * 5);
			List<org.apache.ocean.HttpResponse> httpResponses = nioHttpClient.request(httpRequests, timeoutState);
			for (org.apache.ocean.HttpResponse httpResponse : httpResponses) {
				log.info(httpResponse.toString());
			}
			nioHttpClient.shutdown();
		} catch (Throwable throwable) {
			throwable.printStackTrace();
		}
		System.exit(0);
	}

	public NioHttpClient(int workerCount) throws Exception {
		httpParams = new BasicHttpParams(null);
		httpParams.setIntParameter(CoreConnectionPNames.SO_TIMEOUT, 5000).setIntParameter(CoreConnectionPNames.CONNECTION_TIMEOUT, 10000).setIntParameter(
				CoreConnectionPNames.SOCKET_BUFFER_SIZE, 8 * 1024).setBooleanParameter(CoreConnectionPNames.STALE_CONNECTION_CHECK, false).setBooleanParameter(
				CoreConnectionPNames.TCP_NODELAY, true).setParameter(CoreProtocolPNames.USER_AGENT, "Jakarta-HttpComponents-NIO/1.1");

		ioReactor = new DefaultConnectingIOReactor(workerCount, httpParams);

		BasicHttpProcessor httpProcessor = new BasicHttpProcessor();
		httpProcessor.addInterceptor(new RequestContent());
		httpProcessor.addInterceptor(new RequestTargetHost());
		httpProcessor.addInterceptor(new RequestConnControl());
		httpProcessor.addInterceptor(new RequestUserAgent());
		httpProcessor.addInterceptor(new RequestExpectContinue());

		clientHandler = new BufferingHttpClientHandler(httpProcessor, new CHttpRequestExecutionHandler(), new DefaultConnectionReuseStrategy(), httpParams);
		clientHandler.setEventListener(new EventLogger());
		ioEventDispatch = new DefaultClientIOEventDispatch(clientHandler, httpParams);

		thread = new Thread(new Runnable() {
			public void run() {
				try {
					ioReactor.execute(ioEventDispatch);
				} catch (InterruptedIOException ex) {
					log.severe("Interrupted");
				} catch (IOException e) {
					log.log(Level.SEVERE, "I/O error: " + e.getMessage());
				}
				log.info("Shutdown");
			}

		});
		thread.start();
	}

	public List<org.apache.ocean.HttpResponse> request(List<org.apache.ocean.HttpRequest> httpRequests, TimeoutState timeoutState) throws RequestException, Exception {
		Sessions sessions = new Sessions();
		try {
			for (org.apache.ocean.HttpRequest httpRequest : httpRequests) {
				Session session = execute(httpRequest);
				sessions.add(session);
			}
			while (!sessions.areFinished()) {
				if (timeoutState.notOK()) {
					sessions.cancel();
					break;
				}
				for (Session session : sessions) {
					session.doWait(10);
				}
			}
			// if there were any exceptions the whole request failed
			Exception exception = sessions.getFirstException();
			if (exception != null) throw exception;
			return sessions.getResponses();
		} catch (RequestException requestException) {
			sessions.cancel();
			throw requestException;
		} catch (Exception exception) {
			sessions.cancel();
			log.log(Level.SEVERE, "", exception);
			throw exception;
		} catch (Throwable throwable) {
			sessions.cancel();
			log.log(Level.SEVERE, "", throwable);
			throw new Exception(throwable);
		}
	}

	public class Sessions extends ArrayList<Session> {
		public Exception getFirstException() {
			for (Session session : this) {
				if (session.throwable != null) {
					if (session.throwable instanceof Exception) {
						return (Exception)session.throwable;
					} else {
						return new Exception(session.throwable);
					}
				}
			}
			return null;
		} 
		
		public List<org.apache.ocean.HttpResponse> getResponses() throws RequestException {
			List<org.apache.ocean.HttpResponse> httpResponses = new ArrayList<org.apache.ocean.HttpResponse>();
			for (Session session : this) {
				if (!StringUtils.equals(COMPLETED, session.status)) {
					throw new RequestException(session.attachment, session.getURL(), session.status);
				}
				org.apache.ocean.HttpResponse httpResponse = new org.apache.ocean.HttpResponse();
				httpResponse.bytes = session.responseBytes;
				httpResponse.responseCharSet = session.responseCharset;
				httpResponse.statusCode = session.responseStatusCode;
				httpResponse.attachment = session.attachment;
				httpResponse.headers = session.responseHeaders;
				try {
					httpResponse.url = new URL(session.httpURL.toString());
				} catch (Exception exception) {
					log.log(Level.SEVERE, "", exception);
				}
				httpResponses.add(httpResponse);
			}
			return httpResponses;
		}

		public void cancel() {
			for (Session session : this) {
				session.cancel(null);
			}
		}

		public boolean areFinished() {
			for (Session session : this) {
				if (!session.isFinished)
					return false;
			}
			return true;
		}
	}

	private Session execute(org.apache.ocean.HttpRequest httpRequest) throws Exception {
		HttpURL httpURL = toHttpURL(httpRequest.url);
		if (httpRequest.parameters != null) {
			String queryString = httpRequest.parameters.toEncodedQueryString();
			httpURL.setEscapedQuery(queryString);
		}
		Session session = new Session(httpRequest.attachment, httpURL);
		SessionRequest sessionRequest = ioReactor.connect(new InetSocketAddress(httpURL.getHost(), httpURL.getPort()), null, session, session);
		if (httpRequest.timeout != null) {

		}
		session.attachment = httpRequest.attachment;
		session.sessionRequest = sessionRequest;
		return session;
	}

	private HttpURL toHttpURL(URL url) throws Exception {
		HttpURL httpURL = new HttpURL(url.toString());
		return httpURL;
	}

	public class CHttpRequestExecutionHandler implements HttpRequestExecutionHandler {
		private final static String REQUEST_SENT = "request-sent";
		private final static String RESPONSE_RECEIVED = "response-received";

		public CHttpRequestExecutionHandler() {
			super();
		}

		public void initalizeContext(final HttpContext context, final Object attachment) {
			Session session = (Session) attachment;
			context.setAttribute("session", session);
		}

		public void finalizeContext(final HttpContext context) {
			Object flag = context.getAttribute(RESPONSE_RECEIVED);
			if (flag == null) {
				Session session = (Session) context.getAttribute("session");
				session.finished();
			}
		}

		public HttpRequest submitRequest(final HttpContext context) {
			Session session = (Session) context.getAttribute("session");
			try {
				Object flag = context.getAttribute(REQUEST_SENT);
				if (flag == null) {
					context.setAttribute(REQUEST_SENT, Boolean.TRUE);
					return new BasicHttpRequest("GET", session.httpURL.getPath());
				} else {
					// No new request to submit
					return null;
				}
			} catch (Throwable throwable) {
				log.log(Level.SEVERE, "", throwable);
				session.cancel(throwable);
				return null;
			}
		}

		public void handleResponse(final HttpResponse response, final HttpContext context) {
			HttpEntity httpEntity = response.getEntity();
			Session session = (Session) context.getAttribute("session");
			try {
				Header[] headers = response.getAllHeaders();
				NamedList<String, String> namedListHeaders = new NamedList<String, String>();
				for (Header header : headers) {
					namedListHeaders.add(header.getName(), header.getValue());
				}
				session.responseHeaders = namedListHeaders;
				String charset = EntityUtils.getContentCharSet(httpEntity);
				session.responseBytes = EntityUtils.toByteArray(httpEntity);
				session.responseCharset = charset;
				session.responseHeaders = namedListHeaders;
				StatusLine statusLine = new StatusLine(response.getStatusLine().toString());
				// session.responseStatusLine = statusLine;
				session.responseStatusCode = statusLine.getStatusCode();
			} catch (IOException ioException) {
				log.severe("I/O error: " + ioException.getMessage());
				session.cancel(ioException);
				return;
			}
			context.setAttribute(RESPONSE_RECEIVED, Boolean.TRUE);
			session.finished();
		}
	}

	private class Session implements SessionRequestCallback {
		public HttpURL httpURL;
		ReentrantLock lock = new ReentrantLock();
		Condition finished = lock.newCondition();
		public String status;
		public Object attachment;
		public Throwable throwable;
		SessionRequest sessionRequest;
		public byte[] responseBytes;
		public String responseCharset;
		public int responseStatusCode;
		public NamedList<String, String> responseHeaders;
		boolean isFinished = false;

		public Session(Object attachment, HttpURL httpURL) {
			this.attachment = attachment;
			this.httpURL = httpURL;
		}

		public URL getURL() {
			try {
				return new URL(httpURL.toString());
			} catch (Throwable throwable) {
				return null;
			}
		}

		public void doWait(long timeout) throws InterruptedException {
			if (isFinished)
				return;
			lock.lock();
			try {
				finished.await(timeout, TimeUnit.MILLISECONDS);
			} finally {
				lock.unlock();
			}
		}

		private void cancel(Throwable throwable) {
			if (isFinished)
				return;
			this.throwable = throwable;
			isFinished = true;
			sessionRequest.cancel();
			lock.lock();
			try {
				finished.signalAll();
			} finally {
				lock.unlock();
			}
		}

		private void finished() {
			if (isFinished)
				return;
			IOException ioException = sessionRequest.getException();
			if (ioException != null) {
				throwable = ioException;
			}
			isFinished = true;
			lock.lock();
			try {
				finished.signalAll();
			} finally {
				lock.unlock();
			}
		}

		public void completed(SessionRequest sessionRequest) {
			status = COMPLETED;
		}

		public void failed(SessionRequest sessionRequest) {
			status = FAILED;
			finished();
		}

		public void timeout(SessionRequest sessionRequest) {
			status = TIMEOUT;
			finished();
		}

		public void cancelled(SessionRequest sessionRequest) {
			status = CANCELLED;
			finished();
		}
	}

	public void shutdown() throws Exception {
		ioReactor.shutdown();
	}

	static class EventLogger implements EventListener {
		public void connectionOpen(final NHttpConnection httpConnection) {
			// System.out.println("Connection open: " + conn);
		}

		public void connectionTimeout(final NHttpConnection httpConnection) {
			// System.out.println("Connection timed out: " + conn);
		}

		public void connectionClosed(final NHttpConnection httpConnection) {
			// System.out.println("Connection closed: " + conn);
		}

		public void fatalIOException(final IOException ex, final NHttpConnection httpConnection) {
			// System.err.println("I/O error: " + ex.getMessage());
		}

		public void fatalProtocolException(final HttpException ex, final NHttpConnection httpConnection) {
			// System.err.println("HTTP error: " + ex.getMessage());
		}
	}

	/**
	 * public static class Responses extends HashMap<String, Response> { }
	 * 
	 * public static class Response { }
	 * 
	 * public static class Request { public String id; public URL url; }
	 */
}
